/*
 * SvrTask.cpp
 *
 *  Created on: 2012-7-11
 *      Author: ThinkPad
 */

#include "SvrTask.h"

/**
 * SvrTask class init
 */
SvrTask::SvrTask()
{
    Status = SV_INIT;
    svr_heartbeat_task = TASK_CLOSE;
    pSvrconfig = NULL;

    lock_flag = false;
    default_size = DFLT_MIN_CONN_SIZ;

    heartbeat_buf = NULL;
    heartbeat_buf_offset = 0;
    heartbeat_buf_size = 0;

    task_time_tick = 0;
}
/*
 * SvrTask class free
 */
SvrTask::~SvrTask()
{
	/*free heartbeat buf*/
	if(heartbeat_buf != NULL)
	{
		delete heartbeat_buf;
		heartbeat_buf = NULL;
	}
	/**
	 * first free online task to free pool
	 */
	task_list_lock.Lock();
	{
		delivery_task_node * pfreenode = NULL;
		delivery_task_node * pnode = task_list.get_first();
		while(pnode != NULL)
		{
			pfreenode = pnode;
			pnode = pnode->next;
			task_table.remove(pfreenode->key);
			task_list.remove(pfreenode);
			this->free_node(pfreenode);
		}

		/**
		 * free pool
		 */
		pfreenode = NULL;
		free_list_lock.Lock();
		{
			pnode = free_list.get_first();
			while(pnode != NULL)
			{
				pfreenode = pnode;
				pnode = pnode->next;

				free(pfreenode->plock);
				free(pfreenode);
				pfreenode = NULL;
			}
			free_list_lock.Unlock();
		}

		task_list_lock.Unlock();
	}
}

/**
 * SvrTask init memory
 */
int SvrTask::init(SvrConfig * pconfig,Svr_log * plog,SvrFileManage * psvrfilemanage,int init_task_size)
{
    if(pconfig == NULL)
        return FAILURE;

    this->pSvrconfig = pconfig;
    this->pSvrlog = plog;
    this->pSvrFileUtils = psvrfilemanage;

    default_size = init_task_size;

	if(!task_table.init_table())
		return FAILURE;

	if(!free_list.init_list(false)  ||  !task_list.init_list(true))
		return FAILURE;

	/**
	 * init task pool
	 */
	for(int i=0; i < default_size ; i++)
	{
		delivery_task_node * f_task = new delivery_task_node();
		if(_NULL == f_task)
			return FAILURE;

		free_list.push(f_task);
	}

	/**
	 * init buf pool
	 */
	if(mempool.init_buf() != SUCCESS)
	{
		return FAILURE;
	}

	/*init heartbeat buf*/
	heartbeat_buf = new char[HearBeatBufSize];
	if(heartbeat_buf == NULL)
		return FAILURE;

    return SUCCESS;
}

/*
*svr heartbeat function
*/
uint32_t SvrTask::struct_heartbeat_data(char * buf,size_t buf_len)
{
	uint32_t ret = 0;
	/**
	 * notice delivery status
	 *
	*<?xml version="1.0" encoding="UTF-8"?>
	*<list>
	*<task filename="/1.ts" csize="102400" code="0"/>
	*<task filename="/2.ts" csize="102400" code="0"/>
	*</list>
	 *
	 */
	task_list_lock.Lock();
	{
		memset(this->heartbeat_buf,'\0',HearBeatBufSize);
		this->heartbeat_buf_offset = 0;
		this->heartbeat_buf_size = 0;
		strcpy(this->heartbeat_buf,"<?xml version=\"1.0\" encoding=\"UTF-8\"?><list>");
		delivery_task_node * pnode = task_list.get_first();
		while(pnode != NULL)
		{
			this->heartbeat_buf_size = strlen(this->heartbeat_buf);
			sprintf(this->heartbeat_buf + this->heartbeat_buf_size,"<task filename=\"%s\" csize=\"%llu\" code=\"%d\"/>",
					pnode->filename + this->pSvrconfig->get_root_path_len() , pnode->current_file_size, pnode->errorcode);

			pnode = pnode->next;
		}
		strcat(this->heartbeat_buf,"</list>");
		task_list_lock.Unlock();
	}

	/*copy data to buf*/
	this->heartbeat_buf_size = strlen(this->heartbeat_buf);
	ret =  http::struct_request_header(buf,buf_len,"get.xml",this->pSvrconfig->auth_host,NULL,0,0,this->heartbeat_buf_size);
	size_t clen  = buf_len - ret;
	if(this->heartbeat_buf_size  <= clen)
	{
		clen = this->heartbeat_buf_size;
	}
	if(clen > 0 )
	{
		memcpy(buf + ret, this->heartbeat_buf + this->heartbeat_buf_offset, clen);
		this->heartbeat_buf_offset += clen;
		ret += clen;
	}

	return ret;
}

/**
 * read_heartbeat_data
 */
uint32_t SvrTask::read_heartbeat_data(char * buf,size_t buf_len)
{
	uint32_t ret = 0;
	if(this->heartbeat_buf_offset == this->heartbeat_buf_size)
	{
		return ret;
	}

	size_t clen = this->heartbeat_buf_size - this->heartbeat_buf_offset;
	if(clen >= buf_len)
	{
		clen = buf_len;
	}

	if(clen > 0)
	{
		memcpy(buf, this->heartbeat_buf + this->heartbeat_buf_offset, clen);
		this->heartbeat_buf_offset += clen;
		ret = clen;
	}

	return ret;
}
/*
 * write_heartbeat_data
 */
uint32_t SvrTask::write_heartbeat_data(char * buf , size_t size ,int flag)
{
	uint32_t ret = size;
	if(!flag)
	{
		memset(this->heartbeat_buf,'\0',HearBeatBufSize);
		this->heartbeat_buf_offset = 0;
		this->heartbeat_buf_size = 0;
	}

	if(this->heartbeat_buf_offset + size < HearBeatBufSize)
	{
		memcpy(this->heartbeat_buf + this->heartbeat_buf_offset, buf, size);
		this->heartbeat_buf_offset += size;
	}

	return ret;
}

/*
*task data handle
*/
int SvrTask::handle_heartbeat_data()
{
	if(this->heartbeat_buf_offset <= 0)
		return 0;

	this->heartbeat_buf_size = this->heartbeat_buf_offset;
	this->heartbeat_buf_offset = 0;
    this->pSvrlog->log_fmt(LOG_INFO,"handle_heartbeat_data:%s",this->heartbeat_buf);

    /*check data completed*/
	int httpcode = 0;
	uint64_t dlen = 0;
	size_t hlen = http::parse_response_header(this->heartbeat_buf,this->heartbeat_buf_size,httpcode,dlen);
	if(httpcode != 200
			|| hlen == 0
			|| dlen == 0
			|| dlen + hlen != this->heartbeat_buf_size)
	{
		return -2;
    }

	/**
	 * parse data xml
	 *<?xml version="1.0" encoding="UTF-8"?>
	 *<list>
	 *<task filename="1.ts" size="102400" ip="127.0.0.1" hsize="10240"/>
	 *<task filename="2.ts" size="102400" ip="127.0.0.1" hsize="10240"/>
	 *<task filename="3.ts" size="102400" ip="127.0.0.1" hsize="10240"/>
	 *</list>
	 */
    TiXmlDocument tasklistdoc;
    tasklistdoc.Parse(this->heartbeat_buf + hlen);
    TiXmlHandle docHandle(&tasklistdoc);
	TiXmlElement * xElement = NULL;
	xElement =  docHandle.FirstChildElement("list").FirstChildElement("task").ToElement();
	uint8_t outbuf[GUID_LEN];
	char 	key[GUID_LEN];
	while(xElement != NULL)
	{
		/*
		 * check xml format is ok.
		 * */
		if(xElement->Attribute("filename") == NULL
			||xElement->Attribute("size") == NULL
			|| xElement->Attribute("ip") == NULL
			|| xElement->Attribute("hsize") == NULL)
		{
			break;
		}

		printf("%s,%s,%s,%s\r\n",xElement->Attribute("filename"),xElement->Attribute("size"),xElement->Attribute("ip"),xElement->Attribute("hsize"));

		/*
		 * check task info
		 **/
		const char * pfilename = xElement->Attribute("filename");
		memset(outbuf,0,GUID_LEN);
		memset(key,'\0',GUID_LEN);
		if(pfilename != NULL && strlen(pfilename) > 0)
		{
			md5_sum(outbuf,(uint8_t *)pfilename,strlen(pfilename));
			/*to hex*/
			utils::hexa(key,outbuf,16);
		}
		delivery_task_node * ptask = this->find_task(key);
		if(ptask == NULL)
		{
			this->alloc_node(&ptask);
			if(ptask != NULL)
			{
				strcpy(ptask->filename,this->pSvrconfig->rootpath);
				strcat(ptask->filename,pfilename);
				strcpy(ptask->key,key);
				ptask->filetotalsize = strtoull(xElement->Attribute("size"),NULL,10);
				ptask->remote_ipaddr = inet_addr(xElement->Attribute("ip"));
				ptask->remote_hsize = strtoull(xElement->Attribute("hsize"),NULL,10);
				if(xElement->Attribute("port") != NULL)
				{
					ptask->remote_port = atoi(xElement->Attribute("port"));
				}
				else
				{
					ptask->remote_port = this->pSvrconfig->listen_port;
				}
				ptask->start_time = time_t(NULL);
				ptask->heart_time = time_t(NULL);
				this->add_task(ptask);
			}
			else
			{
				this->pSvrlog->log_fmt(LOG_ERR,"handle_heartbeat_data:alloc_node task error!");
			}
		}
		else
		{
			ptask->remote_ipaddr = inet_addr(xElement->Attribute("ip"));
			ptask->remote_hsize = atoll(xElement->Attribute("hsize"));
			if(xElement->Attribute("port") != NULL)
			{
				ptask->remote_port = atoi(xElement->Attribute("port"));
			}
			else
			{
				ptask->remote_port = this->pSvrconfig->listen_port;
			}
			ptask->heart_time = time_t(NULL);
		}

		xElement = xElement->NextSiblingElement();
	}

    /*hanld end*/
    return SUCCESS;
}

/*
*SvrTask resource pool alloc
*/
int SvrTask::alloc_node(delivery_task_node ** t)
{
	int retValue = FAILURE;
	free_list_lock.Lock();
	{
		if(free_list.size() == 0)
		{
			(*t) = new delivery_task_node();
			if((*t) != NULL)
			{
				retValue = SUCCESS;
			}
		}
		else
		{
			retValue = free_list.alloc(t);
			if(retValue && (*t) != NULL)
			{
				memset((*t)->key,'\0',sizeof((*t)->key));
			}
		}

		free_list_lock.Unlock();
	}
	return retValue;
}

/**
* free task buf
**/
int SvrTask::free_node_buf(delivery_task_node * t)
{
	if(NULL != t)
	{
		buf_chunk * ptr = t->file_data_buf;
		while(ptr != NULL)
		{
			t->file_data_buf = ptr->next;
			this->mempool.free_buf(ptr);
			ptr = t->file_data_buf;
		}
		t->file_data_buf = NULL;
		t->tail = t->file_data_buf;

		/*
		 * close fd
		 * */
		if(t->fd != INVALID_HANDLE_VALUE)
		{
			close(t->fd);
			t->fd = INVALID_HANDLE_VALUE;
		}
	}

	return 0;
}

/*
*SvrTask resource pool free
*/
int SvrTask::free_node(delivery_task_node * t)
{
	int retValue = FAILURE;
	free_list_lock.Lock();
	{
		if(NULL != t)
		{
			t->plock->Lock();
			{
				memset(t->key,'\0',sizeof(t->key));
				memset(t->filename,'\0',sizeof(t->filename));

				t->filetotalsize = 0;
				t->current_file_size = 0;

				t->start_time = time_t(NULL);
				t->heart_time = time_t(NULL);
				t->remote_ipaddr = 0;
				t->remote_port = 0;

				t->status = TASK_CLOSE;
				t->flag = TASK_NONE;
				t->is_completed = false;
				t->is_noticed = false;
				t->errorcode = 0;

				t->key_index = 0;
				t->hash_next = t->hash_prev = NULL;

				/**
				 * free buf,and close file
				 */
				this->free_node_buf(t);

				/*reset completed*/
				t->plock->Unlock();
			}

			/*
			 * push to free list
			 */
			retValue = free_list.push(t);
		}

		free_list_lock.Unlock();
	}
	return retValue;
}

/*
*session resource count
*/
int SvrTask::get_task_count()
{
	return this->task_list.size();
}

int SvrTask::get_free_task_count()
{
	return this->free_list.size();
}

/*
 *struct task request header
 **/
uint32_t SvrTask::struct_task_request_header(const char * key,char * buf,size_t buf_len)
{
	uint32_t ret = 0;
	delivery_task_node * ptask = this->find_task(key);
	if(ptask != NULL)
	{
		ptask->plock->Lock();
		{
			if(!this->create_task_file(ptask))
			{
				this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]struct_task_request_header,create_task_file error!",ptask->key,ptask->filename);
				ptask->plock->Unlock();
				return ret;
			}
			/*check task file  is competed*/
			if(ptask->current_file_size >= ptask->filetotalsize
				|| ptask->current_file_size >= ptask->remote_hsize)
			{
				this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]struct_task_request_header,task file  is competed!",ptask->key,ptask->filename);
				ptask->plock->Unlock();
				return ret;
			}

			char * path = ptask->filename + this->pSvrconfig->get_root_path_len() + 1;
			ret =  http::struct_request_header(buf,buf_len,path,this->pSvrconfig->auth_host,NULL,ptask->current_file_size);

			ptask->plock->Unlock();
		}
	}

	return ret;
}


/**
*write task data
*key:task key
*data: task data
*data_len:task data len
*return :write len,0,write error,-1 task not exist
*/
int SvrTask::write_file_data(const char * key,char * data,size_t data_len)
{
	if(key == NULL || data == NULL  || data_len == 0 || data_len > IO_CHUNK_SIZE)
		return 0;

	int ret = 0;
	delivery_task_node * ptask = this->find_task(key);
	if(ptask != NULL)
	{
		ptask->plock->Lock();
		{
			if(ptask->fd == INVALID_HANDLE_VALUE)
			{
				this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]write file data task errr,fd is null!",ptask->key,ptask->filename);
				ptask->plock->Unlock();
				return ret;
			}
			/*
			 *check task is competed
			 **/
			if(ptask->current_file_size >= ptask->filetotalsize)
			{
				ptask->is_completed = true;
				this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]write file data task completed!",ptask->key,ptask->filename);
				ptask->plock->Unlock();
				return ret;
			}
			/**check header**/
			if(ptask->file_data_buf == NULL)
			{
				this->mempool.alloc_buf(&ptask->file_data_buf);
				if(ptask->file_data_buf  == NULL)
				{
					ptask->plock->Unlock();
					return ret;
				}
				ptask->tail = ptask->file_data_buf;
				ptask->tail->foffset = ptask->current_file_size;
				ptask->buf_count = 1;
			}

			/*
			 *first copy data to tail buf
			 **/
			char * ptr = data;
			size_t freelen = ptask->tail->buf_len - ptask->tail->offset;
			size_t cplen = freelen;
			if(data_len < freelen)
			{
				cplen = data_len;
			}
			if(cplen > 0)
			{
				memcpy(ptask->tail->buffer + ptask->tail->offset,ptr, cplen);
				ptask->tail->offset += cplen;
				ptr = data + cplen;
				cplen = data_len - cplen;
			}
			else
			{
				cplen = data_len;
			}
			/*check is alloc new buf*/
			if(ptask->tail->size == 0 && ptask->tail->offset >= ptask->tail->buf_len)
			{
				/*
				 * set data size
				 **/
				ptask->tail->size = ptask->tail->offset;

				/**
				 * write data to file
				 */
				ssize_t wlen = write(ptask->fd,ptask->tail->buffer,ptask->tail->size);
				if(wlen != (ssize_t)ptask->tail->size)
				{
					this->pSvrlog->log_fmt(LOG_ERR,"[%s][%s]write file data error!",ptask->key,ptask->filename);
					ptask->plock->Unlock();
					return ret;
				}
				else
				{
					ptask->current_file_size += ptask->tail->size;
				}

				/*
				 * check is have completed
				 * */
				if(ptask->current_file_size >= ptask->filetotalsize)
				{
					ptask->is_completed = true;
					this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]write file data task completed!",ptask->key,ptask->filename);
				}
				else
				{
					/*
					 * alloc new task
					 * */
					buf_chunk *  pbuf = NULL;
					this->mempool.alloc_buf(&pbuf);
					if(pbuf != NULL)
					{
						ptask->tail->next = pbuf;
						pbuf->prev = ptask->tail;
						ptask->tail = pbuf;
						ptask->tail->foffset = ptask->current_file_size;
						ptask->buf_count ++;
					}
				}
			}

			/*
			 *second copy last data
			 * */
			if(cplen > 0 && ptask->tail->offset + cplen <= ptask->tail->buf_len)
			{
				memcpy(ptask->tail->buffer + ptask->tail->offset,ptr, cplen);
				ptask->tail->offset += cplen;
			}

			/*
			 * check is have completed
			 * */
			if(!ptask->is_completed && ptask->current_file_size +  ptask->tail->offset >= ptask->filetotalsize)
			{
				/*
				 * task completed
				 **/
				ptask->tail->size = ptask->tail->offset;

				/**
				 * write data to file
				 */
				ssize_t wlen = write(ptask->fd,ptask->tail->buffer,ptask->tail->size);
				if(wlen != (ssize_t)ptask->tail->size)
				{
					this->pSvrlog->log_fmt(LOG_ERR,"[%s][%s]write file data error!",ptask->key,ptask->filename);
					ptask->plock->Unlock();
					return ret;
				}
				else
				{
					ptask->is_completed = true;
					ptask->current_file_size += ptask->tail->size;
					this->pSvrlog->log_fmt(LOG_INFO,"[%s][%s]write file data task completed!",ptask->key,ptask->filename);
				}
			}

			/*check limit buf count*/
			if(ptask->buf_count >= TASK_BUF_LIMT_COUNT)
			{
				/**
				 * free buf
				 */
				buf_chunk * fbuf = ptask->file_data_buf;
				ptask->file_data_buf = fbuf ->next;
				ptask->file_data_buf->prev = NULL;
				this->mempool.free_buf(fbuf);
				ptask->buf_count --;
			}

			/*write ok*/
			ret = data_len;
			ptask->plock->Unlock();
		}
	}
	else
	{
		ret  = -1;
	}

	return ret;
}

/**
 *task run close
 */
int SvrTask::task_run_close(const char * key,int error_code)
{
	if(key == NULL)
		return 0;

	int ret = 0;
	delivery_task_node * ptask = this->find_task(key);
	if(ptask != NULL)
	{
		ptask->plock->Lock();
		{
			ptask->status = TASK_CLOSE;
			ptask->errorcode = error_code;
			ret = 1;

			/*clear buf*/
			this->free_node_buf(ptask);

			ptask->plock->Unlock();
		}
	}

	return ret;
}

/*
*use it must be carefull
*task_list_lock, lock do enter call,
*when you usr completed,then call end_enter
*/
delivery_task_node * SvrTask::begin_enter(void)
{
	delivery_task_node * pnode = NULL;
	task_list_lock.Lock();
	pnode =  task_list.get_first();
	lock_flag = true;
	return pnode;
}

void SvrTask::end_enter(void)
{
	task_list_lock.Unlock();
	lock_flag = false;
}


/**
* init task
*/
bool SvrTask::create_task_file(delivery_task_node * t)
{
	if(t == NULL)
		return false;

	/**
	 * open file by create flag
	 */
	if(t->fd == INVALID_HANDLE_VALUE)
	{
		t->fd = this->pSvrFileUtils->open_file(t->filename,t->current_file_size,true);
	}

	if(t->fd == INVALID_HANDLE_VALUE)
	{
		return false;
	}

	return true;
}

/*
*find task
*key: file name
*/
delivery_task_node* SvrTask::find_task(const char * key)
{
	if(NULL == key || strlen(key)== 0)
		return NULL;

	delivery_task_node * node =  NULL;
	task_list_lock.Lock();
	{
		node =  task_table.find(key);

		task_list_lock.Unlock();
	}

	return node;
}

/*
*add task
*if false,and auto free node
*/
bool SvrTask::add_task(delivery_task_node * t)
{
	bool ret = false;
	task_list_lock.Lock();
	{
		if(NULL != t && strlen(t->key) != 0)
		{
			t->key_index = task_table.add(t,t->key);
			if(t->key_index >= 0)
			{
				ret = task_list.push(t);
			}
			else
			{
				this->free_node(t);
			}
		}
		task_list_lock.Unlock();
	}
	return ret;

}

/*
*remove task
*/
int SvrTask::remove_task(const char * key)
{
	int nRet = FAILURE;
	if(NULL != key && strlen(key) > 0)
	{
		delivery_task_node *p_task = NULL;
		task_list_lock.Lock();
		{
			p_task = task_table.remove(key);
			if(task_list.remove(p_task) && free_node(p_task))
			{
				 nRet =  SUCCESS;
			}

			task_list_lock.Unlock();
		}
	}
	return nRet;
}

/*
* run function
* run
* */
void SvrTask::run(long useconds)
{
	if(task_time_tick * useconds > 1000000)
	{
		task_list_lock.Lock();
		{
			int nspan_time = 0;
			time_t timeCur = time(NULL);
			delivery_task_node * pfreenode = NULL;
			delivery_task_node * pnode = task_list.get_first();
			while(pnode != NULL)
			{
				pfreenode = pnode;
				pnode = pnode->next;
				/*check time out*/
				nspan_time = int(timeCur - pfreenode->heart_time);
				if(nspan_time > TASK_TIME_OUT)
				{
					//this->pSvrlog->log_fmt(LOG_WARN,"file %s cdn task time out!",pfreenode->filename);
				}
				/**
				 * check task is can remove
				 */
				if(pfreenode->is_completed && pfreenode->is_noticed && pfreenode->status == TASK_CLOSE)
				{
					/*remove task*/
					//this->task_table.remove(pfreenode->key);
					/**/
				}
			}

			task_list_lock.Unlock();
		}

		/*
		 * *reset time tick
		 * */
		task_time_tick = 0;
	}

	task_time_tick ++;
}


/**
 * test
 */
void SvrTask::test(const char * filename,uint64_t size,const char * remote_ip,uint64_t remote_size)
{
	if(filename == NULL || remote_ip == NULL)
		return ;

	uint8_t outbuf[GUID_LEN];
	memset(outbuf,0,GUID_LEN);
	md5_sum(outbuf,(uint8_t *)filename,strlen(filename));
	/*to hex*/
	char key[GUID_LEN];
	memset(key,'\0',GUID_LEN);
	utils::hexa(key,outbuf,16);

	delivery_task_node * ptask = this->find_task(key);
	if(ptask == NULL)
	{
		this->alloc_node(&ptask);
		if(ptask != NULL)
		{
			strcpy(ptask->filename,this->pSvrconfig->rootpath);
			strcat(ptask->filename,filename);
			strcpy(ptask->key,key);
			ptask->filetotalsize = size;
			ptask->remote_ipaddr = inet_addr(remote_ip);
			ptask->remote_hsize = remote_size;
			ptask->start_time = time_t(NULL);
			ptask->heart_time = time_t(NULL);
			this->add_task(ptask);
		}
		else
		{
			this->pSvrlog->log_fmt(LOG_ERR,"handle_heartbeat_data:alloc_node task error!");
		}
	}
}
